package com.example.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class GrayMQPushConsumerFactory {

    private static List<GrayMQPushConsumer> consumers;

    static {
        consumers = new ArrayList<>(8);
    }

    public static void addConsumer(GrayMQPushConsumer consumer) {
        synchronized (consumer) {
            consumers.add(consumer);
        }
    }

    public static void shutDownAll() {
        if (!CollectionUtils.isEmpty(consumers)) {
            int successCount = 0;
            for (DefaultMQPushConsumer consumer : consumers) {
                try {
                    consumer.shutdown();
                    ++ successCount;
                    log.info("==>> Mq consumer[{}] is shutting down", consumer.getInstanceName());
                } catch (Exception e) {
                    log.error("Mq consumer shut down error" ,e);
                    consumer.shutdown(); // try again
                }
            }
            log.info("==>> Shut down all mq consumers, totalCount={}, successCount={}", consumers.size(), successCount);
            consumers.clear();
        }
    }
}
